package com.jerome.election.server.handler;

import com.jerome.election.server.ContextUtil;
import com.jerome.election.server.paxos.PaxosMessage;
import com.jerome.election.server.PaxosMethodEnum;
import com.jerome.election.server.PaxosUtil;
import com.jerome.election.storage.ExpiringMap;
import com.jerome.election.sync.CenterServerHandler;
import com.jerome.election.sync.SyncClient;
import com.jerome.election.sync.SyncVo;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.util.Map;

/**
 * 接受请求服务
 * 扮演角色:Acceptor 和 Learner
 *
 * @author Jerome
 * @date 2023/10/18 10:13
 * @version 1.0
 */
public class ServerHandler implements CenterServerHandler {

    @Override
    public void handle(Socket socket) throws IOException {
        // 逐行读取
        BufferedReader httpReader = new BufferedReader(new InputStreamReader(socket.getInputStream(), "UTF-8"));
        PaxosMessage paxosMessage = new PaxosMessage(httpReader.readLine());
        PaxosMethodEnum paxosMethodEnum = PaxosMethodEnum.getByCode(paxosMessage.getMethod());
        switch (paxosMethodEnum){
            case PREPARE:
                if (ContextUtil.getProposeAccepter().prepareProposalNumber(paxosMessage.getProposalNumber())) {
                    // 接受提案
                    System.out.println("Acceptor:Accept proposal:"+paxosMessage);
                    PaxosUtil.write(getWriter(socket),paxosMessage);
                }
                break;
            case ACCEPT:
                if (ContextUtil.getProposeAccepter().acceptProposalNumber(paxosMessage.getProposalNumber())) {
                    // 接受
                    System.out.println("Acceptor:Agree to the proposal:"+paxosMessage);
                    PaxosUtil.write(getWriter(socket),paxosMessage);
                    //通知leaner学习
                    for (Map.Entry<Integer, SyncClient> entry : ContextUtil.getAcceptors().entrySet()) {
                        if (ContextUtil.getNodeMap().get(entry.getKey())!=null){
                            entry.getValue().send(new SyncVo(PaxosMethodEnum.LEARN.getCode(),paxosMessage.getProposalNumber(),paxosMessage.getProposalValue()));
                        }
                    }
                }
                break;
            case LEARN:
                //学习提案
                if (ContextUtil.getLeaner().learnProposal(paxosMessage)){
                    System.out.println("Leaner:Learning Proposal:"+paxosMessage);
                }
                break;
            case LEADER:
                //接收 LEADER 心跳
                ContextUtil.getProposeAccepter().heartbeat(paxosMessage);
                break;
            case PINGPONG:
                //接收心跳
                receiveHeartbeat(socket, paxosMessage);
                break;
            default:
                break;
        }
    }
    /**
     * 接收心跳
     * @param socket
     * @param paxosMessage
     * @author songxulin
     * @date 2023/10/31 15:48
     */
    private void receiveHeartbeat(Socket socket, PaxosMessage paxosMessage) throws IOException {
        int node = Integer.parseInt(paxosMessage.getProposalValue());
        if (ContextUtil.getNodeMap().get(node)==null){
            PaxosMessage leader = ContextUtil.getProposeAccepter().getLeader();
            SyncClient client = ContextUtil.getAcceptors().get(node);
            if (leader!=null&&client!=null){
                client.send(new SyncVo(leader.getMethod(),leader.getProposalNumber(),leader.getProposalValue()));
            }
        }
        PaxosUtil.write(getWriter(socket),new PaxosMessage(paxosMessage.getMethod(),paxosMessage.getProposalNumber(),ContextUtil.getConfig().getNode()+""));
    }

    /**
     * socket 长连接io流
     * Socket:PrintWriter
     */
    private ExpiringMap<Socket, PrintWriter> socketIoMap = new ExpiringMap<>(75000);
    private PrintWriter getWriter(Socket socket) throws IOException {
        PrintWriter writer = socketIoMap.get(socket);
        if (writer ==null){
            writer = new PrintWriter(socket.getOutputStream());
            socketIoMap.put(socket,writer);
        }
        return writer;
    }
}
